package com.caius.xiaohashu.comment.biz.consumer;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.caius.framework.common.util.JsonUtils;
import com.caius.xiaohashu.comment.biz.constant.MQConstants;
import com.caius.xiaohashu.comment.biz.constant.RedisKeyConstants;
import com.caius.xiaohashu.comment.biz.domain.dataobject.CommentDO;
import com.caius.xiaohashu.comment.biz.domain.mapper.CommentDOMapper;
import com.caius.xiaohashu.comment.biz.enums.CommentLevelEnum;
import com.caius.xiaohashu.comment.biz.model.bo.CommentBO;
import com.caius.xiaohashu.comment.biz.model.dto.CountPublishCommentMqDTO;
import com.caius.xiaohashu.comment.biz.model.dto.PublishCommentMqDTO;
import com.caius.xiaohashu.comment.biz.rpc.KeyValueRpcService;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;

import java.util.*;
import java.util.stream.Collectors;

/**
 * @author Caius
 * @description 评论批量写库
 * @since Created in 2025-06-05
 */
@Component
@Slf4j
public class Comment2DBConsumer {

    @Value("${rocketmq.name-server}")
    private String nameservAddr;

    private DefaultMQPushConsumer consumer;

    private RateLimiter rateLimiter = RateLimiter.create(1000);

    @Resource
    private CommentDOMapper commentDOMapper;

    @Resource
    private KeyValueRpcService keyValueRpcService;

    @Resource
    private TransactionTemplate transactionTemplate;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Bean
    public DefaultMQPushConsumer mqPushConsumer() throws MQClientException {
        String group = "xiaohashu_group_" + MQConstants.TOPIC_PUBLISH_COMMENT;

        consumer = new DefaultMQPushConsumer(group);

        consumer.setNamesrvAddr(nameservAddr);

        // 订阅指定的主题，并设置主题的订阅规则（"*" 表示订阅所有标签的消息）
        consumer.subscribe(MQConstants.TOPIC_PUBLISH_COMMENT, "*");

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        consumer.setMessageModel(MessageModel.CLUSTERING);

        consumer.setConsumeMessageBatchMaxSize(30);

        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            List<PublishCommentMqDTO> publishCommentMqDTOS = Lists.newArrayList();

            log.info("==> 本批次消息大小: {}", msgs.size());
            try {
                rateLimiter.acquire();

                for (MessageExt msg : msgs) {
                    String msgJson = new String(msg.getBody());
                    log.info("==> Consumer - Received message: {}", msgJson);
                    // 添加到集合中间
                    publishCommentMqDTOS.add(JsonUtils.parseObject(msgJson , PublishCommentMqDTO.class));
                }

                // 提取所有不为空的评论
                List<Long> replyIds = publishCommentMqDTOS.stream()
                        .filter(publishCommentMqDTO -> Objects.nonNull(publishCommentMqDTO.getCommentId()))
                        .map(PublishCommentMqDTO::getCommentId).toList();

                // 查询所有的评论
                List<CommentDO> replyCommentDOS = null;
                if (CollUtil.isNotEmpty(replyIds)) {
                    replyCommentDOS = commentDOMapper.selectByCommentIds(replyIds);
                }

                // 转换 MAP<ID, COMMENT>
                Map<Long, CommentDO> commentIdAndCommentMap = Maps.newHashMap();
                if (CollUtil.isNotEmpty(replyCommentDOS)) {
                    commentIdAndCommentMap = replyCommentDOS.stream().collect(Collectors.toMap(CommentDO::getId, commentDO -> commentDO));
                }

                // DTO 转 BO
                List<CommentBO> commentBOS = Lists.newArrayList();
                for (PublishCommentMqDTO publishCommentMqDTO : publishCommentMqDTOS) {
                    String imageUrl = publishCommentMqDTO.getImageUrl();
                    CommentBO commentBO = CommentBO.builder()
                            .id(publishCommentMqDTO.getCommentId())
                            .noteId(publishCommentMqDTO.getNoteId())
                            .userId(publishCommentMqDTO.getCreatorId())
                            .isContentEmpty(true) // 默认评论为空
                            .imageUrl(StringUtils.isBlank(imageUrl) ? "" : imageUrl)
                            .level(CommentLevelEnum.ONE.getCode()) // 默认为一级评论
                            .parentId(publishCommentMqDTO.getNoteId()) // 默认所属的笔记 ID
                            .createTime(publishCommentMqDTO.getCreateTime())
                            .updateTime(publishCommentMqDTO.getCreateTime())
                            .isTop(false)
                            .replyTotal(0L)
                            .likeTotal(0L)
                            .replyCommentId(0L)
                            .replyUserId(0L)
                            .build();

                    // 评论内容不为空
                    String content = publishCommentMqDTO.getContent();
                    if (StringUtils.isNotBlank(content)) {
                        commentBO.setContentUuid(UUID.randomUUID().toString()); // 生成内容标识 UUID
                        commentBO.setIsContentEmpty(false);
                        commentBO.setContent(content);
                    }

                     // 设置评论级别、回复用户 ID (reply_user_id)、父评论 ID (parent_id)
                    Long replyCommentId = publishCommentMqDTO.getReplyCommentId();
                    if (Objects.nonNull(replyCommentId)) {
                        CommentDO replyCommentDO = commentIdAndCommentMap.get(replyCommentId);

                        if (Objects.nonNull(replyCommentDO)) {
                            commentBO.setLevel(CommentLevelEnum.TWO.getCode());
                            commentBO.setReplyCommentId(publishCommentMqDTO.getReplyCommentId());
                            // 父评论 ID
                            commentBO.setParentId(replyCommentDO.getId());
                            if (Objects.equals(replyCommentDO.getLevel(), CommentLevelEnum.TWO.getCode())) {
                                commentBO.setParentId(replyCommentDO.getParentId());
                            }
                            // 回复的用户
                            commentBO.setReplyUserId(replyCommentDO.getUserId());
                        }
                    }
                    commentBOS.add(commentBO);
                }
                log.info("## 清洗后的 CommentBOS: {}", JsonUtils.toJsonString(commentBOS));

                Integer insertedRows = transactionTemplate.execute(status -> {
                    try {
                        int count = commentDOMapper.batchInsert(commentBOS);

                        List<CommentBO> commentContentNotEmptyBOS = commentBOS.stream()
                                .filter(commentBO -> Boolean.FALSE.equals(commentBO.getIsContentEmpty()))
                                .toList();

                        if (CollUtil.isNotEmpty(commentContentNotEmptyBOS)) {
                            keyValueRpcService.batchSaveCommentContent(commentContentNotEmptyBOS);
                        }

                        return count;
                    } catch (Exception e) {
                        status.setRollbackOnly();
                        log.error("", e);
                        throw e;
                    }
                });

                // 如果批量插入数据大于 0
                if (Objects.nonNull(insertedRows) && insertedRows > 0) {
                    List<CountPublishCommentMqDTO> countPublishCommentMqDTO = commentBOS.stream()
                            .map(commentBO -> CountPublishCommentMqDTO.builder()
                                    .noteId(commentBO.getNoteId())
                                    .commentId(commentBO.getId())
                                    .level(commentBO.getLevel())
                                    .parentId(commentBO.getParentId())
                                    .build())
                            .toList();

                    // 异步发送计数 MQ
                    Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(countPublishCommentMqDTO))
                            .build();

                    // 同步一级评论到 Redis 热点评论 ZSET 中
                    syncOneLevelComment2RedisZSet(commentBOS);

                    rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_NOTE_COMMENT, message, new SendCallback() {
                        @Override
                        public void onSuccess(SendResult sendResult) {
                            log.info("==> 【计数: 评论发布】MQ 发送成功，SendResult: {}", sendResult);
                        }

                        @Override
                        public void onException(Throwable throwable) {
                            log.error("==> 【计数: 评论发布】MQ 发送异常: ", throwable);
                        }
                    });
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            } catch (Exception e) {
                log.error("", e);
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });

        consumer.start();
        return consumer;
    }

    /**
     * 同步一级评论到 redis 热点评论 ZSet 中
     * @param commentBOS
     */
    private void syncOneLevelComment2RedisZSet(List<CommentBO> commentBOS) {
        Map<Long, List<CommentBO>> commectIdAndBOListMap = commentBOS.stream()
                .filter(commentBO -> Objects.equals(commentBO.getLevel(), CommentLevelEnum.ONE.getCode()))
                .collect(Collectors.groupingBy(CommentBO::getNoteId));

        commectIdAndBOListMap.forEach((noteId, commentBOList) -> {
            String commentListKey = RedisKeyConstants.buildCommentListKey(noteId);

            DefaultRedisScript<Long> script = new DefaultRedisScript<>();

            script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/add_hot_comments.lua")));

            script.setResultType(Long.class);

            List<Object> args = Lists.newArrayList();

            commentBOList.forEach(commentBO -> {
                args.add(commentBO.getId());
                args.add(0);
            });

            redisTemplate.execute(script, Collections.singletonList(commentListKey), args.toArray());
        });
    }

    @PreDestroy
    public void destroy() {
        if (Objects.nonNull(consumer)) {
            try {
                consumer.shutdown();
            } catch (Exception e) {
                log.error("", e);
            }
        }
    }
}
